package com.kafka.provider.controller;

import com.rabbitmq.entity.User;
import net.minidev.json.JSONValue;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.requestreply.RequestReplyFuture;
import org.springframework.kafka.support.SendResult;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @作者: 林江
 * @创建时间: 2024/4/2
 * @功能:
 */
@RestController
public class KafkaProviderController {
    @Autowired
    KafkaTemplate kafkaTemplate;

    //    @Autowired
//    RoutingKafkaTemplate routingKafkaTemplate
    @GetMapping("send/{mark}")
    public String kafkaTemplate(@PathVariable("mark") String mark) {
        if (mark.equals("one")) {
            kafkaTemplate.send("one", "这个是one");
        } else {
            // 发送二进制
//            kafkaTemplate.send("tow", "这个是tow".getBytes(StandardCharsets.UTF_8));
            kafkaTemplate.send("tow", "这个是tow");
        }

        // 当 ProducerFactory 设置了 producerPerThread = true，就需要执行下面的操作。
        // 关闭当前线程的生产者
        kafkaTemplate.getProducerFactory().closeThreadBoundProducer();
        return "success";
    }

    @Autowired
    ReplyingKafkaTemplate replyingKafkaTemplate;

    @GetMapping("reply")
    public String replyingKafkaTemplate() throws ExecutionException, InterruptedException, TimeoutException {
        ProducerRecord record = new ProducerRecord("tow", "测试回复消息的发送");
        // 消费者响应的topic是配置的监听容器中使用的topic(默认也是)
//        record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, "reply".getBytes(StandardCharsets.UTF_8)));

        RequestReplyFuture future = replyingKafkaTemplate.sendAndReceive(record);

        // 发送消息的元数据
        SendResult sendResult = (SendResult) future.getSendFuture().get(5, TimeUnit.SECONDS);
        System.out.println("发送消息：" + sendResult.getRecordMetadata());

        // 接收回复的消息
        ConsumerRecord consumerRecord = (ConsumerRecord) future.get(10, TimeUnit.SECONDS);
        System.out.println(consumerRecord);
        return "success";
    }

    @GetMapping("send/domain")
    public String kafkaTemplate() {
        ProducerRecord record = new ProducerRecord("domain", 0, "linjiang", JSONValue.toJSONString(User.builder().name("林江").age(25).build()));
        kafkaTemplate.send(record);
        return "success";
    }
}
